home *** CD-ROM | disk | FTP | other *** search
/ Chip 2007 January, February, March & April / Chip-Cover-CD-2007-02.iso / Pakiet bezpieczenstwa / mini Pentoo LiveCD 2006.1 / mpentoo-2006.1.iso / livecd.squashfs / usr / lib / python2.4 / test / test_csv.pyc (.txt) < prev    next >
Encoding:
Python Compiled Bytecode  |  2005-10-18  |  32.8 KB  |  1,227 lines
  1. # Source Generated with Decompyle++
  2. # File: in.pyc (Python 2.4)
  3.  
  4. import sys
  5. import os
  6. import unittest
  7. from StringIO import StringIO
  8. import tempfile
  9. import csv
  10. import gc
  11. from test import test_support
  12.  
  13. class Test_Csv(unittest.TestCase):
  14.     '''
  15.     Test the underlying C csv parser in ways that are not appropriate
  16.     from the high level interface. Further tests of this nature are done
  17.     in TestDialectRegistry.
  18.     '''
  19.     
  20.     def test_reader_arg_valid(self):
  21.         self.assertRaises(TypeError, csv.reader)
  22.         self.assertRaises(TypeError, csv.reader, None)
  23.         self.assertRaises(AttributeError, csv.reader, [], bad_attr = 0)
  24.         self.assertRaises(csv.Error, csv.reader, [], 'foo')
  25.         
  26.         class BadClass:
  27.             
  28.             def __init__(self):
  29.                 raise IOError
  30.  
  31.  
  32.         self.assertRaises(IOError, csv.reader, [], BadClass)
  33.         self.assertRaises(TypeError, csv.reader, [], None)
  34.         
  35.         class BadDialect:
  36.             bad_attr = 0
  37.  
  38.         self.assertRaises(AttributeError, csv.reader, [], BadDialect)
  39.  
  40.     
  41.     def test_writer_arg_valid(self):
  42.         self.assertRaises(TypeError, csv.writer)
  43.         self.assertRaises(TypeError, csv.writer, None)
  44.         self.assertRaises(AttributeError, csv.writer, StringIO(), bad_attr = 0)
  45.  
  46.     
  47.     def _test_attrs(self, obj):
  48.         self.assertEqual(obj.dialect.delimiter, ',')
  49.         obj.dialect.delimiter = '\t'
  50.         self.assertEqual(obj.dialect.delimiter, '\t')
  51.         self.assertRaises(TypeError, delattr, obj.dialect, 'delimiter')
  52.         self.assertRaises(TypeError, setattr, obj.dialect, 'lineterminator', None)
  53.         obj.dialect.escapechar = None
  54.         self.assertEqual(obj.dialect.escapechar, None)
  55.         self.assertRaises(TypeError, delattr, obj.dialect, 'quoting')
  56.         self.assertRaises(TypeError, setattr, obj.dialect, 'quoting', None)
  57.         obj.dialect.quoting = csv.QUOTE_MINIMAL
  58.         self.assertEqual(obj.dialect.quoting, csv.QUOTE_MINIMAL)
  59.  
  60.     
  61.     def test_reader_attrs(self):
  62.         self._test_attrs(csv.reader([]))
  63.  
  64.     
  65.     def test_writer_attrs(self):
  66.         self._test_attrs(csv.writer(StringIO()))
  67.  
  68.     
  69.     def _write_test(self, fields, expect, **kwargs):
  70.         (fd, name) = tempfile.mkstemp()
  71.         fileobj = os.fdopen(fd, 'w+b')
  72.         
  73.         try:
  74.             writer = csv.writer(fileobj, **kwargs)
  75.             writer.writerow(fields)
  76.             fileobj.seek(0)
  77.             self.assertEqual(fileobj.read(), expect + writer.dialect.lineterminator)
  78.         finally:
  79.             fileobj.close()
  80.             os.unlink(name)
  81.  
  82.  
  83.     
  84.     def test_write_arg_valid(self):
  85.         self.assertRaises(csv.Error, self._write_test, None, '')
  86.         self._write_test((), '')
  87.         self._write_test([
  88.             None], '""')
  89.         self.assertRaises(csv.Error, self._write_test, [
  90.             None], None, quoting = csv.QUOTE_NONE)
  91.         
  92.         class BadList:
  93.             
  94.             def __len__(self):
  95.                 return 10
  96.  
  97.             
  98.             def __getitem__(self, i):
  99.                 if i > 2:
  100.                     raise IOError
  101.                 
  102.  
  103.  
  104.         self.assertRaises(IOError, self._write_test, BadList(), '')
  105.         
  106.         class BadItem:
  107.             
  108.             def __str__(self):
  109.                 raise IOError
  110.  
  111.  
  112.         self.assertRaises(IOError, self._write_test, [
  113.             BadItem()], '')
  114.  
  115.     
  116.     def test_write_bigfield(self):
  117.         bigstring = 'X' * 50000
  118.         self._write_test([
  119.             bigstring,
  120.             bigstring], '%s,%s' % (bigstring, bigstring))
  121.  
  122.     
  123.     def test_write_quoting(self):
  124.         self._write_test([
  125.             'a',
  126.             '1',
  127.             'p,q'], 'a,1,"p,q"')
  128.         self.assertRaises(csv.Error, self._write_test, [
  129.             'a',
  130.             '1',
  131.             'p,q'], 'a,1,"p,q"', quoting = csv.QUOTE_NONE)
  132.         self._write_test([
  133.             'a',
  134.             '1',
  135.             'p,q'], 'a,1,"p,q"', quoting = csv.QUOTE_MINIMAL)
  136.         self._write_test([
  137.             'a',
  138.             '1',
  139.             'p,q'], '"a",1,"p,q"', quoting = csv.QUOTE_NONNUMERIC)
  140.         self._write_test([
  141.             'a',
  142.             '1',
  143.             'p,q'], '"a","1","p,q"', quoting = csv.QUOTE_ALL)
  144.  
  145.     
  146.     def test_write_escape(self):
  147.         self._write_test([
  148.             'a',
  149.             '1',
  150.             'p,q'], 'a,1,"p,q"', escapechar = '\\')
  151.         self._write_test([
  152.             'a',
  153.             '1',
  154.             'p,q'], 'a,1,p\\,q', escapechar = '\\', quoting = csv.QUOTE_NONE)
  155.  
  156.     
  157.     def test_writerows(self):
  158.         
  159.         class BrokenFile:
  160.             
  161.             def write(self, buf):
  162.                 raise IOError
  163.  
  164.  
  165.         writer = csv.writer(BrokenFile())
  166.         self.assertRaises(IOError, writer.writerows, [
  167.             [
  168.                 'a']])
  169.         (fd, name) = tempfile.mkstemp()
  170.         fileobj = os.fdopen(fd, 'w+b')
  171.         
  172.         try:
  173.             writer = csv.writer(fileobj)
  174.             self.assertRaises(TypeError, writer.writerows, None)
  175.             writer.writerows([
  176.                 [
  177.                     'a',
  178.                     'b'],
  179.                 [
  180.                     'c',
  181.                     'd']])
  182.             fileobj.seek(0)
  183.             self.assertEqual(fileobj.read(), 'a,b\r\nc,d\r\n')
  184.         finally:
  185.             fileobj.close()
  186.             os.unlink(name)
  187.  
  188.  
  189.     
  190.     def _read_test(self, input, expect, **kwargs):
  191.         reader = csv.reader(input, **kwargs)
  192.         result = list(reader)
  193.         self.assertEqual(result, expect)
  194.  
  195.     
  196.     def test_read_oddinputs(self):
  197.         self._read_test([], [])
  198.         self._read_test([
  199.             ''], [
  200.             []])
  201.         self.assertRaises(csv.Error, self._read_test, [
  202.             '"ab"c'], None, strict = 1)
  203.         self.assertRaises(csv.Error, self._read_test, [
  204.             'ab\x00c'], None, strict = 1)
  205.         self._read_test([
  206.             '"ab"c'], [
  207.             [
  208.                 'abc']], doublequote = 0)
  209.  
  210.     
  211.     def test_read_eol(self):
  212.         self._read_test([
  213.             'a,b'], [
  214.             [
  215.                 'a',
  216.                 'b']])
  217.         self._read_test([
  218.             'a,b\n'], [
  219.             [
  220.                 'a',
  221.                 'b']])
  222.         self._read_test([
  223.             'a,b\r\n'], [
  224.             [
  225.                 'a',
  226.                 'b']])
  227.         self._read_test([
  228.             'a,b\r'], [
  229.             [
  230.                 'a',
  231.                 'b']])
  232.         self.assertRaises(csv.Error, self._read_test, [
  233.             'a,b\rc,d'], [])
  234.         self.assertRaises(csv.Error, self._read_test, [
  235.             'a,b\nc,d'], [])
  236.         self.assertRaises(csv.Error, self._read_test, [
  237.             'a,b\r\nc,d'], [])
  238.  
  239.     
  240.     def test_read_escape(self):
  241.         self._read_test([
  242.             'a,\\b,c'], [
  243.             [
  244.                 'a',
  245.                 '\\b',
  246.                 'c']], escapechar = '\\')
  247.         self._read_test([
  248.             'a,b\\,c'], [
  249.             [
  250.                 'a',
  251.                 'b,c']], escapechar = '\\')
  252.         self._read_test([
  253.             'a,"b\\,c"'], [
  254.             [
  255.                 'a',
  256.                 'b,c']], escapechar = '\\')
  257.         self._read_test([
  258.             'a,"b,\\c"'], [
  259.             [
  260.                 'a',
  261.                 'b,\\c']], escapechar = '\\')
  262.         self._read_test([
  263.             'a,"b,c\\""'], [
  264.             [
  265.                 'a',
  266.                 'b,c"']], escapechar = '\\')
  267.         self._read_test([
  268.             'a,"b,c"\\'], [
  269.             [
  270.                 'a',
  271.                 'b,c\\']], escapechar = '\\')
  272.  
  273.     
  274.     def test_read_bigfield(self):
  275.         bigstring = 'X' * 50000
  276.         bigline = '%s,%s' % (bigstring, bigstring)
  277.         self._read_test([
  278.             bigline], [
  279.             [
  280.                 bigstring,
  281.                 bigstring]])
  282.  
  283.  
  284.  
  285. class TestDialectRegistry(unittest.TestCase):
  286.     
  287.     def test_registry_badargs(self):
  288.         self.assertRaises(TypeError, csv.list_dialects, None)
  289.         self.assertRaises(TypeError, csv.get_dialect)
  290.         self.assertRaises(csv.Error, csv.get_dialect, None)
  291.         self.assertRaises(csv.Error, csv.get_dialect, 'nonesuch')
  292.         self.assertRaises(TypeError, csv.unregister_dialect)
  293.         self.assertRaises(csv.Error, csv.unregister_dialect, None)
  294.         self.assertRaises(csv.Error, csv.unregister_dialect, 'nonesuch')
  295.         self.assertRaises(TypeError, csv.register_dialect, None)
  296.         self.assertRaises(TypeError, csv.register_dialect, None, None)
  297.         self.assertRaises(TypeError, csv.register_dialect, 'nonesuch', None)
  298.         
  299.         class bogus:
  300.             
  301.             def __init__(self):
  302.                 raise KeyError
  303.  
  304.  
  305.         self.assertRaises(KeyError, csv.register_dialect, 'nonesuch', bogus)
  306.  
  307.     
  308.     def test_registry(self):
  309.         
  310.         class myexceltsv(csv.excel):
  311.             delimiter = '\t'
  312.  
  313.         name = 'myexceltsv'
  314.         expected_dialects = csv.list_dialects() + [
  315.             name]
  316.         expected_dialects.sort()
  317.         csv.register_dialect(name, myexceltsv)
  318.         
  319.         try:
  320.             self.failUnless(isinstance(csv.get_dialect(name), myexceltsv))
  321.             got_dialects = csv.list_dialects()
  322.             got_dialects.sort()
  323.             self.assertEqual(expected_dialects, got_dialects)
  324.         finally:
  325.             csv.unregister_dialect(name)
  326.  
  327.  
  328.     
  329.     def test_incomplete_dialect(self):
  330.         
  331.         class myexceltsv(csv.Dialect):
  332.             delimiter = '\t'
  333.  
  334.         self.assertRaises(csv.Error, myexceltsv)
  335.  
  336.     
  337.     def test_space_dialect(self):
  338.         
  339.         class space(csv.excel):
  340.             delimiter = ' '
  341.             quoting = csv.QUOTE_NONE
  342.             escapechar = '\\'
  343.  
  344.         (fd, name) = tempfile.mkstemp()
  345.         fileobj = os.fdopen(fd, 'w+b')
  346.         
  347.         try:
  348.             fileobj.write('abc def\nc1ccccc1 benzene\n')
  349.             fileobj.seek(0)
  350.             rdr = csv.reader(fileobj, dialect = space())
  351.             self.assertEqual(rdr.next(), [
  352.                 'abc',
  353.                 'def'])
  354.             self.assertEqual(rdr.next(), [
  355.                 'c1ccccc1',
  356.                 'benzene'])
  357.         finally:
  358.             fileobj.close()
  359.             os.unlink(name)
  360.  
  361.  
  362.     
  363.     def test_dialect_apply(self):
  364.         
  365.         class testA(csv.excel):
  366.             delimiter = '\t'
  367.  
  368.         
  369.         class testB(csv.excel):
  370.             delimiter = ':'
  371.  
  372.         
  373.         class testC(csv.excel):
  374.             delimiter = '|'
  375.  
  376.         csv.register_dialect('testC', testC)
  377.         
  378.         try:
  379.             (fd, name) = tempfile.mkstemp()
  380.             fileobj = os.fdopen(fd, 'w+b')
  381.             
  382.             try:
  383.                 writer = csv.writer(fileobj)
  384.                 writer.writerow([
  385.                     1,
  386.                     2,
  387.                     3])
  388.                 fileobj.seek(0)
  389.                 self.assertEqual(fileobj.read(), '1,2,3\r\n')
  390.             finally:
  391.                 fileobj.close()
  392.                 os.unlink(name)
  393.  
  394.             (fd, name) = tempfile.mkstemp()
  395.             fileobj = os.fdopen(fd, 'w+b')
  396.             
  397.             try:
  398.                 writer = csv.writer(fileobj, testA)
  399.                 writer.writerow([
  400.                     1,
  401.                     2,
  402.                     3])
  403.                 fileobj.seek(0)
  404.                 self.assertEqual(fileobj.read(), '1\t2\t3\r\n')
  405.             finally:
  406.                 fileobj.close()
  407.                 os.unlink(name)
  408.  
  409.             (fd, name) = tempfile.mkstemp()
  410.             fileobj = os.fdopen(fd, 'w+b')
  411.             
  412.             try:
  413.                 writer = csv.writer(fileobj, dialect = testB())
  414.                 writer.writerow([
  415.                     1,
  416.                     2,
  417.                     3])
  418.                 fileobj.seek(0)
  419.                 self.assertEqual(fileobj.read(), '1:2:3\r\n')
  420.             finally:
  421.                 fileobj.close()
  422.                 os.unlink(name)
  423.  
  424.             (fd, name) = tempfile.mkstemp()
  425.             fileobj = os.fdopen(fd, 'w+b')
  426.             
  427.             try:
  428.                 writer = csv.writer(fileobj, dialect = 'testC')
  429.                 writer.writerow([
  430.                     1,
  431.                     2,
  432.                     3])
  433.                 fileobj.seek(0)
  434.                 self.assertEqual(fileobj.read(), '1|2|3\r\n')
  435.             finally:
  436.                 fileobj.close()
  437.                 os.unlink(name)
  438.  
  439.             (fd, name) = tempfile.mkstemp()
  440.             fileobj = os.fdopen(fd, 'w+b')
  441.             
  442.             try:
  443.                 writer = csv.writer(fileobj, dialect = testA, delimiter = ';')
  444.                 writer.writerow([
  445.                     1,
  446.                     2,
  447.                     3])
  448.                 fileobj.seek(0)
  449.                 self.assertEqual(fileobj.read(), '1;2;3\r\n')
  450.             finally:
  451.                 fileobj.close()
  452.                 os.unlink(name)
  453.  
  454.         finally:
  455.             csv.unregister_dialect('testC')
  456.  
  457.  
  458.     
  459.     def test_bad_dialect(self):
  460.         self.assertRaises(AttributeError, csv.reader, [], bad_attr = 0)
  461.         self.assertRaises(TypeError, csv.reader, [], delimiter = None)
  462.         self.assertRaises(TypeError, csv.reader, [], quoting = -1)
  463.         self.assertRaises(TypeError, csv.reader, [], quoting = 100)
  464.  
  465.  
  466.  
  467. class TestCsvBase(unittest.TestCase):
  468.     
  469.     def readerAssertEqual(self, input, expected_result):
  470.         (fd, name) = tempfile.mkstemp()
  471.         fileobj = os.fdopen(fd, 'w+b')
  472.         
  473.         try:
  474.             fileobj.write(input)
  475.             fileobj.seek(0)
  476.             reader = csv.reader(fileobj, dialect = self.dialect)
  477.             fields = list(reader)
  478.             self.assertEqual(fields, expected_result)
  479.         finally:
  480.             fileobj.close()
  481.             os.unlink(name)
  482.  
  483.  
  484.     
  485.     def writerAssertEqual(self, input, expected_result):
  486.         (fd, name) = tempfile.mkstemp()
  487.         fileobj = os.fdopen(fd, 'w+b')
  488.         
  489.         try:
  490.             writer = csv.writer(fileobj, dialect = self.dialect)
  491.             writer.writerows(input)
  492.             fileobj.seek(0)
  493.             self.assertEqual(fileobj.read(), expected_result)
  494.         finally:
  495.             fileobj.close()
  496.             os.unlink(name)
  497.  
  498.  
  499.  
  500.  
  501. class TestDialectExcel(TestCsvBase):
  502.     dialect = 'excel'
  503.     
  504.     def test_single(self):
  505.         self.readerAssertEqual('abc', [
  506.             [
  507.                 'abc']])
  508.  
  509.     
  510.     def test_simple(self):
  511.         self.readerAssertEqual('1,2,3,4,5', [
  512.             [
  513.                 '1',
  514.                 '2',
  515.                 '3',
  516.                 '4',
  517.                 '5']])
  518.  
  519.     
  520.     def test_blankline(self):
  521.         self.readerAssertEqual('', [])
  522.  
  523.     
  524.     def test_empty_fields(self):
  525.         self.readerAssertEqual(',', [
  526.             [
  527.                 '',
  528.                 '']])
  529.  
  530.     
  531.     def test_singlequoted(self):
  532.         self.readerAssertEqual('""', [
  533.             [
  534.                 '']])
  535.  
  536.     
  537.     def test_singlequoted_left_empty(self):
  538.         self.readerAssertEqual('"",', [
  539.             [
  540.                 '',
  541.                 '']])
  542.  
  543.     
  544.     def test_singlequoted_right_empty(self):
  545.         self.readerAssertEqual(',""', [
  546.             [
  547.                 '',
  548.                 '']])
  549.  
  550.     
  551.     def test_single_quoted_quote(self):
  552.         self.readerAssertEqual('""""', [
  553.             [
  554.                 '"']])
  555.  
  556.     
  557.     def test_quoted_quotes(self):
  558.         self.readerAssertEqual('""""""', [
  559.             [
  560.                 '""']])
  561.  
  562.     
  563.     def test_inline_quote(self):
  564.         self.readerAssertEqual('a""b', [
  565.             [
  566.                 'a""b']])
  567.  
  568.     
  569.     def test_inline_quotes(self):
  570.         self.readerAssertEqual('a"b"c', [
  571.             [
  572.                 'a"b"c']])
  573.  
  574.     
  575.     def test_quotes_and_more(self):
  576.         self.readerAssertEqual('"a"b', [
  577.             [
  578.                 'ab']])
  579.  
  580.     
  581.     def test_lone_quote(self):
  582.         self.readerAssertEqual('a"b', [
  583.             [
  584.                 'a"b']])
  585.  
  586.     
  587.     def test_quote_and_quote(self):
  588.         self.readerAssertEqual('"a" "b"', [
  589.             [
  590.                 'a "b"']])
  591.  
  592.     
  593.     def test_space_and_quote(self):
  594.         self.readerAssertEqual(' "a"', [
  595.             [
  596.                 ' "a"']])
  597.  
  598.     
  599.     def test_quoted(self):
  600.         self.readerAssertEqual('1,2,3,"I think, therefore I am",5,6', [
  601.             [
  602.                 '1',
  603.                 '2',
  604.                 '3',
  605.                 'I think, therefore I am',
  606.                 '5',
  607.                 '6']])
  608.  
  609.     
  610.     def test_quoted_quote(self):
  611.         self.readerAssertEqual('1,2,3,"""I see,"" said the blind man","as he picked up his hammer and saw"', [
  612.             [
  613.                 '1',
  614.                 '2',
  615.                 '3',
  616.                 '"I see," said the blind man',
  617.                 'as he picked up his hammer and saw']])
  618.  
  619.     
  620.     def test_quoted_nl(self):
  621.         input = '1,2,3,"""I see,""\nsaid the blind man","as he picked up his\nhammer and saw"\n9,8,7,6'
  622.         self.readerAssertEqual(input, [
  623.             [
  624.                 '1',
  625.                 '2',
  626.                 '3',
  627.                 '"I see,"\nsaid the blind man',
  628.                 'as he picked up his\nhammer and saw'],
  629.             [
  630.                 '9',
  631.                 '8',
  632.                 '7',
  633.                 '6']])
  634.  
  635.     
  636.     def test_dubious_quote(self):
  637.         self.readerAssertEqual('12,12,1",', [
  638.             [
  639.                 '12',
  640.                 '12',
  641.                 '1"',
  642.                 '']])
  643.  
  644.     
  645.     def test_null(self):
  646.         self.writerAssertEqual([], '')
  647.  
  648.     
  649.     def test_single(self):
  650.         self.writerAssertEqual([
  651.             [
  652.                 'abc']], 'abc\r\n')
  653.  
  654.     
  655.     def test_simple(self):
  656.         self.writerAssertEqual([
  657.             [
  658.                 1,
  659.                 2,
  660.                 'abc',
  661.                 3,
  662.                 4]], '1,2,abc,3,4\r\n')
  663.  
  664.     
  665.     def test_quotes(self):
  666.         self.writerAssertEqual([
  667.             [
  668.                 1,
  669.                 2,
  670.                 'a"bc"',
  671.                 3,
  672.                 4]], '1,2,"a""bc""",3,4\r\n')
  673.  
  674.     
  675.     def test_quote_fieldsep(self):
  676.         self.writerAssertEqual([
  677.             [
  678.                 'abc,def']], '"abc,def"\r\n')
  679.  
  680.     
  681.     def test_newlines(self):
  682.         self.writerAssertEqual([
  683.             [
  684.                 1,
  685.                 2,
  686.                 'a\nbc',
  687.                 3,
  688.                 4]], '1,2,"a\nbc",3,4\r\n')
  689.  
  690.  
  691.  
  692. class EscapedExcel(csv.excel):
  693.     quoting = csv.QUOTE_NONE
  694.     escapechar = '\\'
  695.  
  696.  
  697. class TestEscapedExcel(TestCsvBase):
  698.     dialect = EscapedExcel()
  699.     
  700.     def test_escape_fieldsep(self):
  701.         self.writerAssertEqual([
  702.             [
  703.                 'abc,def']], 'abc\\,def\r\n')
  704.  
  705.     
  706.     def test_read_escape_fieldsep(self):
  707.         self.readerAssertEqual('abc\\,def\r\n', [
  708.             [
  709.                 'abc,def']])
  710.  
  711.  
  712.  
  713. class QuotedEscapedExcel(csv.excel):
  714.     quoting = csv.QUOTE_NONNUMERIC
  715.     escapechar = '\\'
  716.  
  717.  
  718. class TestQuotedEscapedExcel(TestCsvBase):
  719.     dialect = QuotedEscapedExcel()
  720.     
  721.     def test_write_escape_fieldsep(self):
  722.         self.writerAssertEqual([
  723.             [
  724.                 'abc,def']], '"abc,def"\r\n')
  725.  
  726.     
  727.     def test_read_escape_fieldsep(self):
  728.         self.readerAssertEqual('"abc\\,def"\r\n', [
  729.             [
  730.                 'abc,def']])
  731.  
  732.  
  733.  
  734. class TestDictFields(unittest.TestCase):
  735.     
  736.     def test_write_simple_dict(self):
  737.         (fd, name) = tempfile.mkstemp()
  738.         fileobj = os.fdopen(fd, 'w+b')
  739.         
  740.         try:
  741.             writer = csv.DictWriter(fileobj, fieldnames = [
  742.                 'f1',
  743.                 'f2',
  744.                 'f3'])
  745.             writer.writerow({
  746.                 'f1': 10,
  747.                 'f3': 'abc' })
  748.             fileobj.seek(0)
  749.             self.assertEqual(fileobj.read(), '10,,abc\r\n')
  750.         finally:
  751.             fileobj.close()
  752.             os.unlink(name)
  753.  
  754.  
  755.     
  756.     def test_write_no_fields(self):
  757.         fileobj = StringIO()
  758.         self.assertRaises(TypeError, csv.DictWriter, fileobj)
  759.  
  760.     
  761.     def test_read_dict_fields(self):
  762.         (fd, name) = tempfile.mkstemp()
  763.         fileobj = os.fdopen(fd, 'w+b')
  764.         
  765.         try:
  766.             fileobj.write('1,2,abc\r\n')
  767.             fileobj.seek(0)
  768.             reader = csv.DictReader(fileobj, fieldnames = [
  769.                 'f1',
  770.                 'f2',
  771.                 'f3'])
  772.             self.assertEqual(reader.next(), {
  773.                 'f1': '1',
  774.                 'f2': '2',
  775.                 'f3': 'abc' })
  776.         finally:
  777.             fileobj.close()
  778.             os.unlink(name)
  779.  
  780.  
  781.     
  782.     def test_read_dict_no_fieldnames(self):
  783.         (fd, name) = tempfile.mkstemp()
  784.         fileobj = os.fdopen(fd, 'w+b')
  785.         
  786.         try:
  787.             fileobj.write('f1,f2,f3\r\n1,2,abc\r\n')
  788.             fileobj.seek(0)
  789.             reader = csv.DictReader(fileobj)
  790.             self.assertEqual(reader.next(), {
  791.                 'f1': '1',
  792.                 'f2': '2',
  793.                 'f3': 'abc' })
  794.         finally:
  795.             fileobj.close()
  796.             os.unlink(name)
  797.  
  798.  
  799.     
  800.     def test_read_long(self):
  801.         (fd, name) = tempfile.mkstemp()
  802.         fileobj = os.fdopen(fd, 'w+b')
  803.         
  804.         try:
  805.             fileobj.write('1,2,abc,4,5,6\r\n')
  806.             fileobj.seek(0)
  807.             reader = csv.DictReader(fileobj, fieldnames = [
  808.                 'f1',
  809.                 'f2'])
  810.             self.assertEqual(reader.next(), {
  811.                 'f1': '1',
  812.                 'f2': '2',
  813.                 None: [
  814.                     'abc',
  815.                     '4',
  816.                     '5',
  817.                     '6'] })
  818.         finally:
  819.             fileobj.close()
  820.             os.unlink(name)
  821.  
  822.  
  823.     
  824.     def test_read_long_with_rest(self):
  825.         (fd, name) = tempfile.mkstemp()
  826.         fileobj = os.fdopen(fd, 'w+b')
  827.         
  828.         try:
  829.             fileobj.write('1,2,abc,4,5,6\r\n')
  830.             fileobj.seek(0)
  831.             reader = csv.DictReader(fileobj, fieldnames = [
  832.                 'f1',
  833.                 'f2'], restkey = '_rest')
  834.             self.assertEqual(reader.next(), {
  835.                 'f1': '1',
  836.                 'f2': '2',
  837.                 '_rest': [
  838.                     'abc',
  839.                     '4',
  840.                     '5',
  841.                     '6'] })
  842.         finally:
  843.             fileobj.close()
  844.             os.unlink(name)
  845.  
  846.  
  847.     
  848.     def test_read_long_with_rest_no_fieldnames(self):
  849.         (fd, name) = tempfile.mkstemp()
  850.         fileobj = os.fdopen(fd, 'w+b')
  851.         
  852.         try:
  853.             fileobj.write('f1,f2\r\n1,2,abc,4,5,6\r\n')
  854.             fileobj.seek(0)
  855.             reader = csv.DictReader(fileobj, restkey = '_rest')
  856.             self.assertEqual(reader.next(), {
  857.                 'f1': '1',
  858.                 'f2': '2',
  859.                 '_rest': [
  860.                     'abc',
  861.                     '4',
  862.                     '5',
  863.                     '6'] })
  864.         finally:
  865.             fileobj.close()
  866.             os.unlink(name)
  867.  
  868.  
  869.     
  870.     def test_read_short(self):
  871.         (fd, name) = tempfile.mkstemp()
  872.         fileobj = os.fdopen(fd, 'w+b')
  873.         
  874.         try:
  875.             fileobj.write('1,2,abc,4,5,6\r\n1,2,abc\r\n')
  876.             fileobj.seek(0)
  877.             reader = csv.DictReader(fileobj, fieldnames = '1 2 3 4 5 6'.split(), restval = 'DEFAULT')
  878.             self.assertEqual(reader.next(), {
  879.                 '1': '1',
  880.                 '2': '2',
  881.                 '3': 'abc',
  882.                 '4': '4',
  883.                 '5': '5',
  884.                 '6': '6' })
  885.             self.assertEqual(reader.next(), {
  886.                 '1': '1',
  887.                 '2': '2',
  888.                 '3': 'abc',
  889.                 '4': 'DEFAULT',
  890.                 '5': 'DEFAULT',
  891.                 '6': 'DEFAULT' })
  892.         finally:
  893.             fileobj.close()
  894.             os.unlink(name)
  895.  
  896.  
  897.     
  898.     def test_read_multi(self):
  899.         sample = [
  900.             '2147483648,43.0e12,17,abc,def\r\n',
  901.             '147483648,43.0e2,17,abc,def\r\n',
  902.             '47483648,43.0,170,abc,def\r\n']
  903.         reader = csv.DictReader(sample, fieldnames = 'i1 float i2 s1 s2'.split())
  904.         self.assertEqual(reader.next(), {
  905.             'i1': '2147483648',
  906.             'float': '43.0e12',
  907.             'i2': '17',
  908.             's1': 'abc',
  909.             's2': 'def' })
  910.  
  911.     
  912.     def test_read_with_blanks(self):
  913.         reader = csv.DictReader([
  914.             '1,2,abc,4,5,6\r\n',
  915.             '\r\n',
  916.             '1,2,abc,4,5,6\r\n'], fieldnames = '1 2 3 4 5 6'.split())
  917.         self.assertEqual(reader.next(), {
  918.             '1': '1',
  919.             '2': '2',
  920.             '3': 'abc',
  921.             '4': '4',
  922.             '5': '5',
  923.             '6': '6' })
  924.         self.assertEqual(reader.next(), {
  925.             '1': '1',
  926.             '2': '2',
  927.             '3': 'abc',
  928.             '4': '4',
  929.             '5': '5',
  930.             '6': '6' })
  931.  
  932.     
  933.     def test_read_semi_sep(self):
  934.         reader = csv.DictReader([
  935.             '1;2;abc;4;5;6\r\n'], fieldnames = '1 2 3 4 5 6'.split(), delimiter = ';')
  936.         self.assertEqual(reader.next(), {
  937.             '1': '1',
  938.             '2': '2',
  939.             '3': 'abc',
  940.             '4': '4',
  941.             '5': '5',
  942.             '6': '6' })
  943.  
  944.  
  945.  
  946. class TestArrayWrites(unittest.TestCase):
  947.     
  948.     def test_int_write(self):
  949.         import array
  950.         contents = [ 20 - i for i in range(20) ]
  951.         a = array.array('i', contents)
  952.         (fd, name) = tempfile.mkstemp()
  953.         fileobj = os.fdopen(fd, 'w+b')
  954.         
  955.         try:
  956.             writer = csv.writer(fileobj, dialect = 'excel')
  957.             writer.writerow(a)
  958.             expected = []([ str(i) for i in a ]) + '\r\n'
  959.             fileobj.seek(0)
  960.             self.assertEqual(fileobj.read(), expected)
  961.         finally:
  962.             fileobj.close()
  963.             os.unlink(name)
  964.  
  965.  
  966.     
  967.     def test_double_write(self):
  968.         import array
  969.         contents = [ (20 - i) * 0.10000000000000001 for i in range(20) ]
  970.         a = array.array('d', contents)
  971.         (fd, name) = tempfile.mkstemp()
  972.         fileobj = os.fdopen(fd, 'w+b')
  973.         
  974.         try:
  975.             writer = csv.writer(fileobj, dialect = 'excel')
  976.             writer.writerow(a)
  977.             expected = []([ str(i) for i in a ]) + '\r\n'
  978.             fileobj.seek(0)
  979.             self.assertEqual(fileobj.read(), expected)
  980.         finally:
  981.             fileobj.close()
  982.             os.unlink(name)
  983.  
  984.  
  985.     
  986.     def test_float_write(self):
  987.         import array
  988.         contents = [ (20 - i) * 0.10000000000000001 for i in range(20) ]
  989.         a = array.array('f', contents)
  990.         (fd, name) = tempfile.mkstemp()
  991.         fileobj = os.fdopen(fd, 'w+b')
  992.         
  993.         try:
  994.             writer = csv.writer(fileobj, dialect = 'excel')
  995.             writer.writerow(a)
  996.             expected = []([ str(i) for i in a ]) + '\r\n'
  997.             fileobj.seek(0)
  998.             self.assertEqual(fileobj.read(), expected)
  999.         finally:
  1000.             fileobj.close()
  1001.             os.unlink(name)
  1002.  
  1003.  
  1004.     
  1005.     def test_char_write(self):
  1006.         import array
  1007.         import string
  1008.         a = array.array('c', string.letters)
  1009.         (fd, name) = tempfile.mkstemp()
  1010.         fileobj = os.fdopen(fd, 'w+b')
  1011.         
  1012.         try:
  1013.             writer = csv.writer(fileobj, dialect = 'excel')
  1014.             writer.writerow(a)
  1015.             expected = ','.join(a) + '\r\n'
  1016.             fileobj.seek(0)
  1017.             self.assertEqual(fileobj.read(), expected)
  1018.         finally:
  1019.             fileobj.close()
  1020.             os.unlink(name)
  1021.  
  1022.  
  1023.  
  1024.  
  1025. class TestDialectValidity(unittest.TestCase):
  1026.     
  1027.     def test_quoting(self):
  1028.         
  1029.         class mydialect(csv.Dialect):
  1030.             delimiter = ';'
  1031.             escapechar = '\\'
  1032.             doublequote = False
  1033.             skipinitialspace = True
  1034.             lineterminator = '\r\n'
  1035.             quoting = csv.QUOTE_NONE
  1036.  
  1037.         d = mydialect()
  1038.         mydialect.quoting = None
  1039.         self.assertRaises(csv.Error, mydialect)
  1040.         mydialect.quoting = csv.QUOTE_NONE
  1041.         mydialect.escapechar = None
  1042.         self.assertRaises(csv.Error, mydialect)
  1043.         mydialect.doublequote = True
  1044.         mydialect.quoting = csv.QUOTE_ALL
  1045.         mydialect.quotechar = '"'
  1046.         d = mydialect()
  1047.         mydialect.quotechar = "''"
  1048.         self.assertRaises(csv.Error, mydialect)
  1049.         mydialect.quotechar = 4
  1050.         self.assertRaises(csv.Error, mydialect)
  1051.  
  1052.     
  1053.     def test_delimiter(self):
  1054.         
  1055.         class mydialect(csv.Dialect):
  1056.             delimiter = ';'
  1057.             escapechar = '\\'
  1058.             doublequote = False
  1059.             skipinitialspace = True
  1060.             lineterminator = '\r\n'
  1061.             quoting = csv.QUOTE_NONE
  1062.  
  1063.         d = mydialect()
  1064.         mydialect.delimiter = ':::'
  1065.         self.assertRaises(csv.Error, mydialect)
  1066.         mydialect.delimiter = 4
  1067.         self.assertRaises(csv.Error, mydialect)
  1068.  
  1069.     
  1070.     def test_lineterminator(self):
  1071.         
  1072.         class mydialect(csv.Dialect):
  1073.             delimiter = ';'
  1074.             escapechar = '\\'
  1075.             doublequote = False
  1076.             skipinitialspace = True
  1077.             lineterminator = '\r\n'
  1078.             quoting = csv.QUOTE_NONE
  1079.  
  1080.         d = mydialect()
  1081.         mydialect.lineterminator = ':::'
  1082.         d = mydialect()
  1083.         mydialect.lineterminator = 4
  1084.         self.assertRaises(csv.Error, mydialect)
  1085.  
  1086.  
  1087.  
  1088. class TestSniffer(unittest.TestCase):
  1089.     sample1 = "Harry's, Arlington Heights, IL, 2/1/03, Kimi Hayes\nShark City, Glendale Heights, IL, 12/28/02, Prezence\nTommy's Place, Blue Island, IL, 12/28/02, Blue Sunday/White Crow\nStonecutters Seafood and Chop House, Lemont, IL, 12/19/02, Week Back\n"
  1090.     sample2 = "'Harry''s':'Arlington Heights':'IL':'2/1/03':'Kimi Hayes'\n'Shark City':'Glendale Heights':'IL':'12/28/02':'Prezence'\n'Tommy''s Place':'Blue Island':'IL':'12/28/02':'Blue Sunday/White Crow'\n'Stonecutters Seafood and Chop House':'Lemont':'IL':'12/19/02':'Week Back'\n"
  1091.     header = '"venue","city","state","date","performers"\n'
  1092.     sample3 = '05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03\n05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03\n05/05/03?05/05/03?05/05/03?05/05/03?05/05/03?05/05/03\n'
  1093.     sample4 = '2147483648;43.0e12;17;abc;def\n147483648;43.0e2;17;abc;def\n47483648;43.0;170;abc;def\n'
  1094.     
  1095.     def test_has_header(self):
  1096.         sniffer = csv.Sniffer()
  1097.         self.assertEqual(sniffer.has_header(self.sample1), False)
  1098.         self.assertEqual(sniffer.has_header(self.header + self.sample1), True)
  1099.  
  1100.     
  1101.     def test_sniff(self):
  1102.         sniffer = csv.Sniffer()
  1103.         dialect = sniffer.sniff(self.sample1)
  1104.         self.assertEqual(dialect.delimiter, ',')
  1105.         self.assertEqual(dialect.quotechar, '"')
  1106.         self.assertEqual(dialect.skipinitialspace, True)
  1107.         dialect = sniffer.sniff(self.sample2)
  1108.         self.assertEqual(dialect.delimiter, ':')
  1109.         self.assertEqual(dialect.quotechar, "'")
  1110.         self.assertEqual(dialect.skipinitialspace, False)
  1111.  
  1112.     
  1113.     def test_delimiters(self):
  1114.         sniffer = csv.Sniffer()
  1115.         dialect = sniffer.sniff(self.sample3)
  1116.         self.assertEqual(dialect.delimiter, '0')
  1117.         dialect = sniffer.sniff(self.sample3, delimiters = '?,')
  1118.         self.assertEqual(dialect.delimiter, '?')
  1119.         dialect = sniffer.sniff(self.sample3, delimiters = '/,')
  1120.         self.assertEqual(dialect.delimiter, '/')
  1121.         dialect = sniffer.sniff(self.sample4)
  1122.         self.assertEqual(dialect.delimiter, ';')
  1123.  
  1124.  
  1125. if not hasattr(sys, 'gettotalrefcount'):
  1126.     if test_support.verbose:
  1127.         print '*** skipping leakage tests ***'
  1128.     
  1129. else:
  1130.     
  1131.     class NUL:
  1132.         
  1133.         def write(s, *args):
  1134.             pass
  1135.  
  1136.         writelines = write
  1137.  
  1138.     
  1139.     class TestLeaks(unittest.TestCase):
  1140.         
  1141.         def test_create_read(self):
  1142.             delta = 0
  1143.             lastrc = sys.gettotalrefcount()
  1144.             for i in xrange(20):
  1145.                 gc.collect()
  1146.                 self.assertEqual(gc.garbage, [])
  1147.                 rc = sys.gettotalrefcount()
  1148.                 csv.reader([
  1149.                     'a,b,c\r\n'])
  1150.                 csv.reader([
  1151.                     'a,b,c\r\n'])
  1152.                 csv.reader([
  1153.                     'a,b,c\r\n'])
  1154.                 delta = rc - lastrc
  1155.                 lastrc = rc
  1156.             
  1157.             self.assertEqual(delta < 3, True)
  1158.  
  1159.         
  1160.         def test_create_write(self):
  1161.             delta = 0
  1162.             lastrc = sys.gettotalrefcount()
  1163.             s = NUL()
  1164.             for i in xrange(20):
  1165.                 gc.collect()
  1166.                 self.assertEqual(gc.garbage, [])
  1167.                 rc = sys.gettotalrefcount()
  1168.                 csv.writer(s)
  1169.                 csv.writer(s)
  1170.                 csv.writer(s)
  1171.                 delta = rc - lastrc
  1172.                 lastrc = rc
  1173.             
  1174.             self.assertEqual(delta < 3, True)
  1175.  
  1176.         
  1177.         def test_read(self):
  1178.             delta = 0
  1179.             rows = [
  1180.                 'a,b,c\r\n'] * 5
  1181.             lastrc = sys.gettotalrefcount()
  1182.             for i in xrange(20):
  1183.                 gc.collect()
  1184.                 self.assertEqual(gc.garbage, [])
  1185.                 rc = sys.gettotalrefcount()
  1186.                 rdr = csv.reader(rows)
  1187.                 for row in rdr:
  1188.                     pass
  1189.                 
  1190.                 delta = rc - lastrc
  1191.                 lastrc = rc
  1192.             
  1193.             self.assertEqual(delta < 5, True)
  1194.  
  1195.         
  1196.         def test_write(self):
  1197.             delta = 0
  1198.             rows = [
  1199.                 [
  1200.                     1,
  1201.                     2,
  1202.                     3]] * 5
  1203.             s = NUL()
  1204.             lastrc = sys.gettotalrefcount()
  1205.             for i in xrange(20):
  1206.                 gc.collect()
  1207.                 self.assertEqual(gc.garbage, [])
  1208.                 rc = sys.gettotalrefcount()
  1209.                 writer = csv.writer(s)
  1210.                 for row in rows:
  1211.                     writer.writerow(row)
  1212.                 
  1213.                 delta = rc - lastrc
  1214.                 lastrc = rc
  1215.             
  1216.             self.assertEqual(delta < 5, True)
  1217.  
  1218.  
  1219.  
  1220. def test_main():
  1221.     mod = sys.modules[__name__]
  1222.     [](*_[1])
  1223.  
  1224. if __name__ == '__main__':
  1225.     test_main()
  1226.  
  1227.